package com.bungarus.busi;

import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.BungarusServer;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageStatus;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.apache.log4j.Logger;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Created by tang on 2018/11/18.
 */
public class Remedy {
    Logger logger = Logger.getLogger(Remedy.class);

    protected static final int WORK_THREAD = 2;
    protected static final long DELAY_TIME = 3;
    protected static final long PERIOD = 10;
    protected static final long ACK_TIMEOUT = 30 * 60 * 1000;

    protected static boolean started = false;

    protected Map<String, InnerClient> clients ;
    protected final DefaultEventExecutorGroup executorGroup = new DefaultEventExecutorGroup(WORK_THREAD);

    public static void start(BungarusServer server0) throws Exception{
        if(!started) {
            started = true;
            new Remedy().run(server0);
        }
    }
    /**
     * init the remedy
     */
    protected void init(BungarusServer server){
        clients = server.getInnerClients();
    }

    /**
     * main logic for each schedule
     */
    protected void process(){
        DischargeQueue queue = Discharger.getQueueInstance();
        Remedy.this.retry0(queue);
    }

    protected void run(BungarusServer server) throws Exception {
        init(server);
        executorGroup.scheduleAtFixedRate(() -> {
            process();
            logger.info("Remedy is running...");
        }
        ,DELAY_TIME ,PERIOD ,TimeUnit.SECONDS);
        logger.info("Remedy is running...");
    }

    protected void retry0(DischargeQueue queue) {
        logger.debug("Size of discharge queue: " + queue.size());
        if(queue.size() <= 0 || clients.isEmpty()){
            return;
        }
        ConcurrentHashMap.KeySetView<Long, MessageProtobuf.Message> keys = queue.keys();
        for(Long l: keys){
            MessageProtobuf.Message noAckMsg = queue.getFromQueue(l);
            if((System.currentTimeMillis() - noAckMsg.getHeader().getSentTime()) >= ACK_TIMEOUT) {
                MessageProtobuf.Extension extension = noAckMsg.getHeader().getExtensions(0);
                if(null != extension && extension.getKey().equalsIgnoreCase(MessageBuilder.TAG)) {
                    String tag = extension.getValue();
                    try {
                        // if no ack when timeout, remove the message first, and put it into the queue again
                        if(MessageStatus.REMEDY == noAckMsg.getHeader().getReturnCode()) {
                            clients.get(tag).saveTryFailedMessage(noAckMsg);
                        }
                        else {
                            clients.get(tag).sendMessage(MessageBuilder.buildRemedyMessage(noAckMsg), tag);
                        }
                        queue.removeFromQueue(l);
                    }
                    catch (BungarusClientException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}
